package com.shujia.flink.core

import org.apache.flink.streaming.api.scala._

object Demo3Parallelism {
  def main(args: Array[String]): Unit = {

    /**
     * 并行度设置
     * 1、再提交任务时设置 -p 12   (推荐使用方式，灵活，不需要重新打包)  （优先级最低）
     * 2、再代码中统一设置env.setParallelism(2)
     * 3、每一个算子可以单独设置并行度   （优先级最高）
     *
     * 并行度设置原则
     * flink做实时计算一班数据量看的每秒的数据量
     * flink的每个并行度每秒可以处理的数据量
     * 1、非聚合计算时---单并行度 100000/s左右
     * 2、聚合计算时 -- 单并行度  20000/s  左右
     *
     *
     * 举例
     * 每天会产生10TB数据， 应该给多少并行度
     * 10TB   --> 100亿条数据  --> 11万/s * 2   -- 10-50个并行度
     */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //env.setParallelism(2)

    val linesDS: DataStream[String] = env
      .socketTextStream("master", 8888)

    val wordsDS: DataStream[String] = linesDS.flatMap(_.split(","))

    val kvDS: DataStream[(String, Int)] = wordsDS.map(word => (word, 1))

    val keyBHyDS: KeyedStream[(String, Int), String] = kvDS.keyBy(kv => kv._1)

    val countDS: DataStream[(String, Int)] = keyBHyDS
      .sum(1)
      .setParallelism(3) //每一个算子单独设置并行度

    countDS.print()

    env.execute()

  }
}
